package com.atguigu.edu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.bean.TradeSourceOrderBeean;
import com.atguigu.edu.util.KafkaUtil;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import java.time.ZoneId;

/**
 * @author zzw
 * @create 2022/9/5 20:04
 */
public class DwsTradeOrderSourceWindow {
    public static void main(String[] args) {
        // TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // TODO 2 设置状态后端
        /*
        env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
        env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME", "atguigu");
        env.setStateBackend(new HashMapStateBackend());
        */

        // 设置时区的方法
        tableEnv.getConfig().setLocalTimeZone(ZoneId.of("GMT+8"));

        // TODO 3　读取DWD层dwd_trade_order_detail 和dwd_traffic_unique_visitor_detail主题数据
        String topicName = "dwd_trade_order_pre_process";
        String uvTopic = "dwd_traffic_unique_visitor_detail";
        String groupId = "dws_trade_order_source_window";
        DataStreamSource<String> orderSourceStream = env.addSource(KafkaUtil.getKafkaConsumer(topicName, groupId));
        DataStreamSource<String> uvDS = env.addSource(KafkaUtil.getKafkaConsumer(uvTopic, groupId));

        // TODO 3 转换数据类型 + 过滤不完整的数据
        SingleOutputStreamOperator<JSONObject> jsonObjStream = orderSourceStream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                String userId = jsonObject.getString("user_id");
                String type = jsonObject.getString("type");
                if (userId != null && "insert".equals(type)) {
                    out.collect(jsonObject);
                }
            }
        });

        // TODO 5 根据来源sc分组
        KeyedStream<JSONObject, String> keyedStream = jsonObjStream.keyBy(JSONObject -> JSONObject.getString("sc"));
        keyedStream.flatMap(new FlatMapFunction<JSONObject, TradeSourceOrderBeean>() {
            @Override
            public void flatMap(JSONObject value, Collector<TradeSourceOrderBeean> out) throws Exception {

            }
        })

        // TODO 6 转换uvDS数据类型
        uvDS.map(new MapFunction<String, TradeSourceOrderBeean>() {
            @Override
            public TradeSourceOrderBeean map(String value) throws Exception {
                return TradeSourceOrderBeean.builder()
                        .uvCt(1L)
                        .build();
            }
        })



        //

    }
}
